package com.shujia.flink.table

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.Types
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.descriptors._
import org.apache.flink.types.Row

object Demo3tableONKafka {
  def main(args: Array[String]): Unit = {

    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv = StreamTableEnvironment.create(streamEnv)


    //读取kafka数据注册成一张表
    tableEnv
      .connect(
        new Kafka()
          .version("universal")
          .topic("student")
          .startFromEarliest()
          .property("zookeeper.connect", "node1:2181,node2:2181,node3:2181")
          .property("bootstrap.servers", "node1:9092")
          .property("group.id", "asdasdadf"))
      .withFormat(new Json().deriveSchema())
      .withSchema(new Schema()
        .field("id", Types.STRING)
        .field("name", Types.STRING)
        .field("age", Types.LONG)
        .field("gender", Types.STRING)
        .field("clazz", Types.STRING)
      ).inAppendMode()
      .registerTableSource("student")

    val table = tableEnv.scan("student")

    tableEnv.toAppendStream[Row](table).print()

    streamEnv.execute()

  }
}
